'''
@Company: TWL
@Author: xue jian
@Email: xuejian@kanzhun.com
@Date: 2020-04-23 16:49:46
'''
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka import TopicPartition
import time
import json
bootstrap_servers=['172.21.210.125:9092', '172.21.210.96:9092', '172.21.210.102:9092', '172.21.210.78:9092', '172.21.210.54:9092']
producer = KafkaProducer(bootstrap_servers=bootstrap_servers, compression_type='snappy')
produce_topic = 'boss.de_two.recall_trace_log.phoenix' #boss.de_two.recall_trace_log.phoenix

def deal_line(line):
    line = line.split(' ')
    # return line
    # re_m = {}
    info = {'rclTP':'vector_flash_recall', 'query':''}
    millis = str(int(round(time.time() * 1000)))
    source = 'f1_rec'
    bg = '1'
    action = 'vector_flash_recall'
    session_id = ''
    # print(len(line))
    if len(line) != 20:
        return False, ''
    else:
        if line[14] == 'The' and line[15] == 'whole' and line[16] == 'process' and line[17] == 'takes':
            # print('haha')
            uid = ''
            jepid = ''
            if line[9][:3] == 'uid':
                uid = line[9][4:-1]
            if line[10][:5] == 'jepId':
                jepid = line[10][6:-1]
            if line[13][:5] == 'limit':
                info['count'] = line[13][6:-1]
            if line[11][:4] == 'city':
                info['city'] = line[11][5:-1]
            if line[12][:8] == 'position':
                info['position'] = line[12][9:-1]
            info['Tcost'] = line[18]
            msg = uid + '\t' + jepid + '\t' + bg + '\t' + session_id +'\t' + millis + '\t' \
                + action + '\t' + json.dumps(info) +'\t' + source
            return True, msg 
        else:
            return False, ''

            


while True:
    line = input()
    test, msg = deal_line(line)
    if test:
        # print(msg)
        # print(len(msg))
        try:
            producer.send(produce_topic, msg.encode(encoding='utf-8')) 
        except:
            print('send error!!!')
            continue
    else:
        # print(1, msg)
        continue
